﻿using LJ.Abstractions.Config;
using LJ.EventBus.HornedSerpent.Transport.Abstractions;
using LJ.EventBus.HornedSerpent.Transport.Abstractions.Exceptions;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.HornedSerpent.Internal;

public class EventBusSubscribe(IEventBusHandlerManager eventBusHandlerManager, IEventBusMessageSerializer eventBusMessageSerializer,
    IEventBusConsumerClientFactory eventBusConsumerClientFactory, IEventBusMessageListener eventBusMessageListener,
    IConfigProvider configProvider,
    ILogger<EventBusSubscribe> logger) : IEventBusSubscribe
{
    private readonly TimeSpan _pollingDelay = TimeSpan.FromSeconds(1);

    private readonly IEventBusHandlerManager _eventBusHandlerManager = eventBusHandlerManager;
    private readonly IEventBusMessageSerializer _eventBusMessageSerializer = eventBusMessageSerializer;
    private readonly IEventBusMessageListener _eventBusMessageListener = eventBusMessageListener;

    private readonly IEventBusConsumerClientFactory _eventBusConsumerClientFactory = eventBusConsumerClientFactory;
    private readonly EventBusConfig _eventBusConfig = configProvider.Get<EventBusConfig>();
    private readonly ILogger _logger = logger;

    public Task Start(CancellationToken stoppingToken)
    {
        // 加载所有的事件处理类
        var eventBusHandlerDescriptors = _eventBusHandlerManager.GetEventBusHandlerDescriptors();

        // 注册事件订阅
        foreach (var eventBusHandlerDescriptor in eventBusHandlerDescriptors)
        {
            try
            {
                // ReSharper disable once ConvertToUsingDeclaration
                using var client = _eventBusConsumerClientFactory.Create(eventBusHandlerDescriptor.Group, eventBusHandlerDescriptor.Topic);
                client.OnLogCallback = WriteLog;
                client.FetchTopic();
            }
            catch (EventBusBrokerConnectionException e)
            {
                _logger.LogError(e, "BrokerConnectionException:{Message}", e.Message);
                return Task.CompletedTask;
            }

            _ = Task.Factory.StartNew(async () =>
            {
                try
                {
                    // ReSharper disable once ConvertToUsingDeclaration
                    using var client = _eventBusConsumerClientFactory.Create(eventBusHandlerDescriptor.Group, eventBusHandlerDescriptor.Topic);

                    RegisterMessageProcessor(client, stoppingToken);

                    client.Subscribe();

                   await  client.ListeningAsync(_pollingDelay, stoppingToken);
                }
                catch (OperationCanceledException)
                {
                    //ignore
                }
                catch (EventBusBrokerConnectionException e)
                {
                    _logger.LogError(e, "BrokerConnectionException:{Message}", e.Message);
                }
                catch (Exception e)
                {
                    _logger.LogError(e, "Exception:{Message}", e.Message);
                }
            }, stoppingToken, TaskCreationOptions.LongRunning, TaskScheduler.Default);
        }
        return Task.CompletedTask;
    }

    private void RegisterMessageProcessor(IEventBusConsumerClient client, CancellationToken cancellationToken)
    {
        client.OnLogCallback = WriteLog;
        client.OnDataCallback = async (transportData, sender) =>
        {
            try
            {
                var name = transportData.GetName();

                var eventBusHandlerDescriptor = _eventBusHandlerManager.GetEventBusHandlerDescriptor(name);

                
                var message = new EventBusMessage(transportData.Headers,
                   _eventBusMessageSerializer.Deserialize(_eventBusMessageSerializer.ToJson(transportData.Body.Span.ToArray()), eventBusHandlerDescriptor.EventDataType));

                _logger.LogDebug("Received message. id:{messageId}, name: {name}", transportData.GetId(), name);

                // 执行消息监听处理
                if (await _eventBusMessageListener.OnReceiveAsync(message, cancellationToken))
                {
                    // 只有监听处理成功才提交偏移量,否则不处理即可
                    client.Commit(sender);
                }

            }
            catch (Exception e)
            {
                _logger.LogError(e, "An exception occurred when process received message. Message:'{Message}'.", transportData);
                if (_eventBusConfig.SkipNonRetriableError)
                {
                    client.Commit(sender);
                }
                else
                {
                    client.Reject(sender);
                }
            }
        };
    }

    private void WriteLog(EventBusTransportLog log)
    {
        switch (log.Type)
        {
            case EventBusTransportLogType.ConsumeError:
                _logger.LogError("Kafka client consume error. -->{reason} ", log.Reason);
                break;
            case EventBusTransportLogType.ConsumeRetries:
                _logger.LogWarning("Kafka client consume exception, retying... -->{reason} ", log.Reason);
                break;
            case EventBusTransportLogType.ServerConnError:
                _logger.LogCritical("Kafka server connection error. -->{reason} ", log.Reason);
                break;
            default:
                throw new ArgumentOutOfRangeException(nameof(log));
        }
    }

}
